package ods_kafka

import (
	"fmt"
	"github.com/Shopify/sarama"
	"server/global"
)

type OdsKafka struct {
	Name      string `json:"name" form:"name" gorm:"column:name;comment:名称;size:200;"`
	Dsn       string `json:"dsn" form:"dsn" gorm:"column:dsn;comment:dsn;size:200;"`
	Topic     string `json:"topic" form:"topic" gorm:"column:topic;comment:主题;size:200;"`
	Partition string `json:"partition" form:"partition" gorm:"column:partition;comment:分区;size:20;"`
	Bsm       string `json:"bsm" form:"bsm" gorm:"column:bsm;comment:标识码;size:100;"`
	Offset    string `json:"offset" form:"offset" gorm:"column:offset;comment:偏移;size:50;"`


	DbName     string `json:"dbName" form:"dbName" gorm:"column:dbName;comment:数据库名;size:100;"`
	TbName     string `json:"tbName" form:"tbName" gorm:"column:charSet;comment:数据表名;size:50;"`
	FldMap     string `json:"fldMap" form:"fldMap" gorm:"column:fldMap;comment:字段映射;size:255;"`
	TestResult string `json:"testResult" form:"testResult" gorm:"column:testResult;comment:测试结果;size:255;"`
	Enable     bool   `json:"enable" form:"enable" gorm:"column:enable;comment:是否启用;size:255;"`
	Ready      string `json:"ready" form:"ready" gorm:"column:ready;comment:配置检查;size:32;"`
	Start bool `json:"start" form:"start" gorm:"column:start;comment:是否启动;size:32;"`
	Running string `json:"running" gorm:"-"`

	global.BaseModel
}

func (m *OdsKafka) TableName() string {
	return "ods_kafka"
}

func (m *OdsKafka) Save() error {
	err := global.SysDB.Save(&m).Error
	return err
}

func (m *OdsKafka) CheckReady() {
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true

	consumer, err := sarama.NewConsumer([]string{m.Dsn}, config)
	if err != nil {
		return
	}

	partitionList, err := consumer.Partitions(m.Topic)
	if err != nil {
		return
	}
	if len(partitionList) > 0 {
		m.Ready = "连接正常"
		return
	}
	return
}

func (m *OdsKafka) SetBsm() error {
	if m.ID <= 0 {
		return fmt.Errorf("更新错误,数据ID为0")
	}
	m.Bsm = fmt.Sprintf("kfk/%s/%s/%d", m.Dsn, m.Topic, m.ID)
	err := m.Save()
	if err != nil {
		return err
	}
	return nil
}
